package trouger.dmt;

import java.util.ArrayList;

import org.apache.hadoop.mapred.JobConf;

/**
 * The common base class of {@link JoinTableRunner} and {@link SelfAggregationRunner}.
 *
 */
public abstract class JoinableRunnerBase
extends BaseMapredRunner<JoinableMapper, JoinableReducer>{
	
	protected ArrayList<String> aggregationOperation = null;
	
	/**
	 * Set an aggregation operation on a result field.
	 * @param resultField the index(one based) of the result field to which the aggregation should be applied 
	 * @param aggregationOperatorClass a class that implements the {@link AggregationOperator} interface
	 * @param resumeOnError if set to true, the task will continue running when the aggregation operator throws a {@link RuntimeException} on accepting one value,
	 * and a log is printed to STDERR. if set to false, any exception will cause task failure.
	 */
	public void setAggregationOperator(
			int resultField,
			Class<? extends AggregationOperator> aggregationOperatorClass,
			boolean resumeOnError){
		if (aggregationOperation == null){
			aggregationOperation = new ArrayList<String>();
		}
		aggregationOperation.add(aggregationOperatorClass.getName() + "\t" + resumeOnError+ "\t" + resultField);
	}
	/**
	 * Set an aggregation operation on a result field.
	 * Calling this method is equal to calling setAggregationOperator(resultField, aggregationOperatorClass, false);
	 * @param resultField the index(one based) of the result field to which the aggregation should be applied
	 * @param aggregationOperatorClass a class that implements the {@link AggregationOperator} interface
	 */
	public void setAggregationOperator(int resultField, Class<? extends AggregationOperator> aggregationOperatorClass){
		setAggregationOperator(resultField, aggregationOperatorClass, false);
	}
	
	/**
	 * A more convenient version of setResultFields.
	 * This method provides a SQL-like interface for specifying result fields with aggregation operations.
	 * @param resultFieldsEx can be a traditional result-fields string, like "1.1 1.2 2.3", 
	 * or a SQL-like string, e.g. "1.1,count(1.2),avr(2.3)". For more details, look up the toolkit manual.
	 */
	public void setResultFieldsEx(String resultFieldsEx){
		StringBuffer resultFieldsBuf = new StringBuffer();
		int fieldCounter = 0;
		String[] seg = resultFieldsEx.split("[\\s]*[\\s,][\\s]*");
		for (String field : seg){
			fieldCounter++;
			if (field.length() == 0){
				throw new IllegalArgumentException("empty field is not allowed");
			}else if (field.matches("[0-9]+\\.[0-9]+")){
				if (resultFieldsBuf.length() > 0){
					resultFieldsBuf.append(" ");
				}
				resultFieldsBuf.append(field);
			}else if (field.matches("[a-zA-Z_]\\w*\\([0-9]+\\.[0-9]+\\)")){
				String func = field.substring(0, field.indexOf('('));
				Class<? extends AggregationOperator> opClass = PredefinedAggregations.getOperatorClassByAlias(func);
				if (opClass == null){
					throw new IllegalArgumentException("function not found: " + func);
				}
				String actualField = field.substring(field.indexOf('(') + 1, field.indexOf(')'));
				if (resultFieldsBuf.length() > 0){
					resultFieldsBuf.append(" ");
				}
				resultFieldsBuf.append(actualField);
				this.setAggregationOperator(fieldCounter, opClass, true);
			}else{
				throw new IllegalArgumentException("wrong format in resultFieldsEx");
			}
		}
		this.setResultFields(resultFieldsBuf.toString());
	}
	
	@Override 
	protected void configureJob(JobConf conf){
		super.configureJob(conf);
		conf.setOutputValueGroupingComparator(JoinableGroupingComparator.class);

		if (aggregationOperation != null){
			BaseMapredUtils.setConfigStrings(conf, 
					JoinableReducer.AGGREGATION_OPERATIONS_PARAM_NAME, 
					aggregationOperation);
		}
	}

}
